package com.nokia.position;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimeZone;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import com.twitter.elephantbird.mapreduce.input.MultiInputFormat;
import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;
import com.nokia.position.PositionProto.Position;

public class ReadUtil {

	public static class LzoMapper extends
			Mapper<LongWritable, ProtobufWritable<Position>, Text, Text> {
		@Override
		protected void map(LongWritable key, ProtobufWritable<Position> value,
				Context context) throws IOException, InterruptedException {

			Position position = value.get();
			Date dateObj = new Date(position.getCoordinatesTimestamp());

			SimpleDateFormat parser = new SimpleDateFormat("yyyy/MM/dd/HH");
			parser.setTimeZone(TimeZone.getTimeZone("UTC"));
			String partitionDate = parser.format(dateObj);

			
			String delim = context.getConfiguration().get("delim");
			if (delim == null) {
					delim = ",";
			} else if (delim.equals("tab") | delim.equals("tsv")) {
				delim = "\t";
			} else if (delim.equals("csv")) {
				delim = ",";
			}

			if (delim.equals("verbose") | delim.equals("v")) {
				context.write(
						null,
						new Text(
								"Partition Date: ["
										+ partitionDate
										+ "]"
										+ "\t"
										+ "Major Version: ["
										+ (position.hasMajorVersion() ? position
												.getMajorVersion() : "")
										+ "]"
										+ "\t"
										+ "Minor Version: ["
										+ (position.hasMinorVersion() ? position
												.getMinorVersion() : "")
										+ "]"
										+ "\t"
										+ "Time Stamp: ["
										+ (position.hasCoordinatesTimestamp() ? position
												.getCoordinatesTimestamp() : "")
										+ "]"
										+ "\t"
										+ "Latitude: ["
										+ (position.hasLatitude() ? position
												.getLatitude() : "")
										+ "]"
										+ "\t"
										+ "Longitude: ["
										+ (position.hasLongitude() ? position
												.getLongitude() : "")
										+ "]"
										+ "\t"
										+ "Altitude: ["
										+ (position.hasAltitude() ? position
												.getAltitude() : "")
										+ "]"
										+ "\t"
										+ "Haccuracy: ["
										+ (position.hasHaccuracy() ? position
												.getHaccuracy() : "")
										+ "]"
										+ "\t"
										+ "Vaccuracy: ["
										+ (position.hasVaccuracy() ? position
												.getVaccuracy() : "")
										+ "]"
										+ "\t"
										+ "Network Type:  ["
										+ (position.hasNetworkType() ? position
												.getNetworkType() : "")
										+ "]"
										+ "\t"
										+ "MCC: ["
										+ (position.hasServingMccSystemID() ? position
												.getServingMccSystemID() : "")
										+ "]"
										+ "\t"
										+ "MNC: ["
										+ (position.hasServingMncNetworkID() ? position
												.getServingMncNetworkID() : "")
										+ "]"
										+ "\t"
										+ "Home MCC: ["
										+ (position.hasHomeMccSystemID() ? position
												.getHomeMccSystemID() : "")
										+ "]"
										+ "\t"
										+ "Home MNC: ["
										+ (position.hasHomeMncNetworkID() ? position
												.getHomeMncNetworkID() : "")
										+ "]"
										+ "\t"
										+ "Provided Measurements:  ["
										+ (position.hasProvidedMeasurements() ? position
												.getProvidedMeasurements() : "")
										+ "]"
										+ "\t"
										+ "Used Measurements: ["
										+ (position.hasUsedMeasurements() ? position
												.getUsedMeasurements() : "")
										+ "]"
										+ "\t"
										+ "Protocol Type: ["
										+ (position.hasProtocolType() ? position
												.getProtocolType() : "")
										+ "]"
										+ "\t"
										+ "Protocol Version: ["
										+ (position.hasProtocolVersion() ? position
												.getProtocolVersion() : "")
										+ "]"
										+ "\t"
										+ "Protocol Major Version: ["
										+ (position.hasProtocolVersionMajor() ? position
												.getProtocolVersionMajor() : "")
										+ "]"
										+ "\t"
										+ "Protocol Minor Version: ["
										+ (position.hasProtocolVersionMinor() ? position
												.getProtocolVersionMinor() : "")
										+ "]"
										+ "\t"
										+ "Manufacturer: ["
										+ (position.hasManufacturer() ? position
												.getManufacturer() : "")
										+ "]"
										+ "\t"
										+ "Client Type: ["
										+ (position.hasClientType() ? position
												.getClientType() : "")
										+ "]"
										+ "\t"
										+ "Status: ["
										+ (position.hasStatus() ? position
												.getStatus() : "")
										+ "]"
										+ "\t"
										+ "Error Code: ["
										+ (position.hasErrorCode() ? position
												.getErrorCode() : "")
										+ "]"
										+ "\t"
										+ "ModelId: ["
										+ (position.hasModelId() ? position
												.getModelId() : "")
										+ "]"
										+ "\t"
										+ "Client Name: ["
										+ (position.hasClientName() ? position
												.getClientName() : "")
										+ "]"
										+ "\t"
										+ "FQDN: ["
										+ (position.hasFqdn() ? position
												.getFqdn() : "")
										+ "]"
										+ "\t"
										+ "SIM2 Network Type: ["
										+ (position.hasSim2NetworkType() ? position
												.getSim2NetworkType() : "")
										+ "]"
										+ "\t"
										+ "SIM2 MCC: ["
										+ (position.hasSim2ServingMccSystemID() ? position
												.getSim2ServingMccSystemID()
												: "")
										+ "]"
										+ "\t"
										+ "SIM2 MNC: ["
										+ (position
												.hasSim2ServingMncNetworkID() ? position
												.getSim2ServingMncNetworkID()
												: "")
										+ "]"
										+ "\t"
										+ "SIM2 Home MCC: ["
										+ (position.hasSim2HomeMccSystemID() ? position
												.getSim2HomeMccSystemID() : "")
										+ "]"
										+ "\t"
										+ "SIM2 Home MNC: ["
										+ (position.hasSim2HomeMncNetworkID() ? position
												.getSim2HomeMncNetworkID() : "")
										+ "]"
										+ "\t"
										+ "SIM2 Provided Measurements: ["
										+ (position
												.hasSim2ProvidedMeasurements() ? position
												.getSim2ProvidedMeasurements()
												: "")
										+ "]"
										+ "\t"
										+ "SIM2 Used Measurements: ["
										+ (position.hasSim2UsedMeasurements() ? position
												.getSim2UsedMeasurements() : "")
										+ "]"));
			} else {
				context.write(
						null,
						new Text(
								partitionDate
										+ delim
										+ (position.hasMajorVersion() ? position
												.getMajorVersion() : "")
										+ delim
										+ (position.hasMinorVersion() ? position
												.getMinorVersion() : "")
										+ delim
										+ (position.hasCoordinatesTimestamp() ? position
												.getCoordinatesTimestamp() : "")
										+ delim
										+ (position.hasLatitude() ? position
												.getLatitude() : "")
										+ delim
										+ (position.hasLongitude() ? position
												.getLongitude() : "")
										+ delim
										+ (position.hasAltitude() ? position
												.getAltitude() : "")
										+ delim
										+ (position.hasHaccuracy() ? position
												.getHaccuracy() : "")
										+ delim
										+ (position.hasVaccuracy() ? position
												.getVaccuracy() : "")
										+ delim
										+ (position.hasNetworkType() ? position
												.getNetworkType() : "")
										+ delim
										+ (position.hasServingMccSystemID() ? position
												.getServingMccSystemID() : "")
										+ delim
										+ (position.hasServingMncNetworkID() ? position
												.getServingMncNetworkID() : "")
										+ delim
										+ (position.hasHomeMccSystemID() ? position
												.getHomeMccSystemID() : "")
										+ delim
										+ (position.hasHomeMncNetworkID() ? position
												.getHomeMncNetworkID() : "")
										+ delim
										+ (position.hasProvidedMeasurements() ? position
												.getProvidedMeasurements() : "")
										+ delim
										+ (position.hasUsedMeasurements() ? position
												.getUsedMeasurements() : "")
										+ delim
										+ (position.hasProtocolType() ? position
												.getProtocolType() : "")
										+ delim
										+ (position.hasProtocolVersion() ? position
												.getProtocolVersion() : "")
										+ delim
										+ (position.hasProtocolVersionMajor() ? position
												.getProtocolVersionMajor() : "")
										+ delim
										+ (position.hasProtocolVersionMinor() ? position
												.getProtocolVersionMinor() : "")
										+ delim
										+ (position.hasManufacturer() ? position
												.getManufacturer() : "")
										+ delim
										+ (position.hasClientType() ? position
												.getClientType() : "")
										+ delim
										+ (position.hasStatus() ? position
												.getStatus() : "")
										+ delim
										+ (position.hasErrorCode() ? position
												.getErrorCode() : "")
										+ delim
										+ (position.hasModelId() ? position
												.getModelId() : "")
										+ delim
										+ (position.hasClientName() ? position
												.getClientName() : "")
										+ delim
										+ (position.hasFqdn() ? position
												.getFqdn() : "")
										+ delim
										+ (position.hasSim2NetworkType() ? position
												.getSim2NetworkType() : "")
										+ delim
										+ (position.hasSim2ServingMccSystemID() ? position
												.getSim2ServingMccSystemID()
												: "")
										+ delim
										+ (position
												.hasSim2ServingMncNetworkID() ? position
												.getSim2ServingMncNetworkID()
												: "")
										+ delim
										+ (position.hasSim2HomeMccSystemID() ? position
												.getSim2HomeMccSystemID() : "")
										+ delim
										+ (position.hasSim2HomeMncNetworkID() ? position
												.getSim2HomeMncNetworkID() : "")
										+ delim
										+ (position
												.hasSim2ProvidedMeasurements() ? position
												.getSim2ProvidedMeasurements()
												: "")
										+ delim
										+ (position.hasSim2UsedMeasurements() ? position
												.getSim2UsedMeasurements() : "")
										));
			}
		}
	}

	int runLzoToText(String[] args, Configuration conf) throws Exception {
		Job job = new Job(conf);
		job.setJobName("Positioning: LzoB64Line to Text");

		job.setJarByClass(getClass());
		job.setMapperClass(LzoMapper.class);
		job.setNumReduceTasks(0);

		// input format is same for both B64Line or block:
		MultiInputFormat.setInputFormatClass(Position.class, job);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		args = new GenericOptionsParser(conf, args).getRemainingArgs();
		ReadUtil runner = new ReadUtil();

		if (args.length != 2) {
			System.out.println("Usage: hadoop jar path/to/this.jar "
					+ runner.getClass() + " [-Ddelim=verbose|tsv|csv|<string> default=csv] <input dir> <output dir>");
			System.exit(1);
		}

		System.exit(runner.runLzoToText(args, conf));

	}
}
